package hbase.tools;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.NamespaceNotFoundException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;

public class HbaseUtils {

    /**
     * 判断 Namespace 是否存在
     * @param admin
     * @param nsName
     * @return
     */
    public static boolean isNamespaceExists(Admin admin,String nsName){
        boolean flag = true;
        try {
            admin.getNamespaceDescriptor(nsName);
        } catch (NamespaceNotFoundException e){
            flag = false;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return flag;
    }

    /**
     * 判断表是否存在
     * @param admin
     * @param tableName 只需传字符串类型参数即可
     * @return
     */
    public static boolean isTableExists(Admin admin,String tableName){
        try {
            return admin.tableExists(TableName.valueOf(tableName));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 判断表是否不存在
     * @param admin
     * @param tableName 只需传字符串类型参数即可
     * @return
     */
    public static boolean isTableNotExists(Admin admin,String tableName){
        try {
            return !admin.tableExists(TableName.valueOf(tableName));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 新建 TableDescriptor 对象
     * @param tableName
     * @return
     */
    public static TableDescriptorBuilder.ModifyableTableDescriptor createTableDescriptor(String tableName){
        return new TableDescriptorBuilder.ModifyableTableDescriptor(TableName.valueOf(tableName));
    }

    /**
     * 获取已有对 TableDescriptor 对象
     * @param admin
     * @param tableName
     * @return
     */
    public static TableDescriptorBuilder.ModifyableTableDescriptor getTableDescriptor(Admin admin,String tableName){
        try {
            return (TableDescriptorBuilder.ModifyableTableDescriptor)admin.getDescriptor(TableName.valueOf(tableName));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 新建 ColumnFamilyDescriptor 对象
     * @param cfName
     * @return
     */
    public static ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor createCFDescriptor(String cfName){
        return new ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor(Bytes.toBytes(cfName));
    }

    /**
     * 在确定表一定存在的情况下，判断是否有数据
     * @param table
     * @return
     */
    public static boolean isDataExists(Table table){

        boolean flag = false;
        try {
            ResultScanner results = table.getScanner(new Scan());
            Iterator<Result> resIte = results.iterator();
            if(resIte.hasNext()){
                flag = true;
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        return flag;
    }

    /**
     * 在确定表一定存在的情况下，判断是否有数据
     * @param table
     * @return
     */
    public static boolean isDataNotExists(Table table){

        boolean flag = false;
        try {
            ResultScanner results = table.getScanner(new Scan());
            Iterator<Result> resIte = results.iterator();
            if(!resIte.hasNext()){
                flag = true;
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        return flag;
    }

    /**
     * 不确定表是否存在的情况下，同时判断表和数据是否都存在
     * @param admin
     * @param tableName
     * @return
     */
    public static boolean isTableAndDataExists(Admin admin,Table table,String tableName){
        if(isTableExists(admin,tableName)){
            return isDataExists(table);
        }
        return false;
    }

    /**
     * 输出result中的所有信息
     * @param result
     */
    public static void showCellInfo(Result result){
        //获取每行内容的所有的单元格
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            System.out.println("rowKey:"+ new String(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength()));
            System.out.println("columnFamily:"+ new String(cell.getRowArray(),cell.getFamilyOffset(),cell.getFamilyLength()));
            System.out.println("column:"+ new String(CellUtil.cloneQualifier(cell)));
            System.out.println("value:"+ new String(CellUtil.cloneValue(cell)));
            System.out.println("timestamp:"+cell.getTimestamp());
            System.out.println("=====================================================");
        }
    }

    /**
     * 关闭hbase连接
     * @param connection
     */
    public static void closed(Connection connection){
        closed(connection,null,null);
    }

    /**
     * 关闭hbase连接和admin对象
     * @param connection
     * @param admin
     */
    public static void closed(Connection connection,Admin admin){
        closed(connection,admin,null);
    }

    /**
     * 关闭hbase连接和table对象
     * @param connection
     * @param table
     */
    public static void closed(Connection connection,Table table){
        closed(connection,null,table);
    }

    /**
     * 关闭hbase连接和Admin/table对象
     * @param connection
     * @param admin
     */
    public static void closed(Connection connection, Admin admin,Table table){
        try {
            if(table!=null){
                table.close();
            }
            if(admin!=null){
                admin.close();
            }
            if(connection!=null){
                connection.close();
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}
